package dataStream;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.ArrayList;
import java.util.List;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2021/10/12
 */
public class SplitFunc {
    public static void main(String[] args) throws Exception {
        List<Person> data = new ArrayList<>();
        data.add( new Person("man", 20));
        data.add( new Person("man", 18));
        data.add( new Person("man", 16));
        data.add( new Person("woman", 20));
        data.add( new Person("woman", 18));
        data.add( new Person("woman", 16));
        data.add( new Person("woman", 15));

        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Person> dataStream = env.fromCollection(data);

        // 注意实例后面的花括号
        final OutputTag<Person> man = new OutputTag<Person>("1"){};
        final OutputTag<Person> woman = new OutputTag<Person>("0"){};

        SingleOutputStreamOperator<Person> sideOutPutData = dataStream.process(new ProcessFunction<Person, Person>() {
            @Override
            public void processElement(Person person, Context context, Collector<Person> collector) throws Exception {
                String gender = person.getGender();
                if ("man".equals(gender)) {
                    context.output(man, person);
                } else {
                    context.output(woman, person);
                }
            }
        });



        DataStream<Person> sideManOutput = sideOutPutData.getSideOutput(man);
        DataStream<Person> sidewomanOutput = sideOutPutData.getSideOutput(woman);

        sideManOutput.print();

        env.execute();



    }
}
